package org.anno.chunkdb.backend;

import com.google.common.base.Preconditions;

import org.anno.chunkdb.Metadata;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Consumer;
import java.util.function.IntFunction;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.Stream;

/**
 * A bunch of consecutive rows, which are identified simply by an int, which is interpreted by each
 * column to return a value.
 */
public class Chunk implements Table<FlyRow> {

  private final int size;
  private final List<Column<FlyRow>> columnList;

  /**
   * @param size       count of rows.
   * @param columnList columns, where the data is stored for each row.
   */
  public Chunk(int size, List<Column<FlyRow>> columnList) {
    this.size = size;
    this.columnList = columnList;
  }

  @Override
  public List<Column<FlyRow>> getColumns() {
    return columnList;
  }

  @Override
  public Stream<FlyRow> getAll() {
    return IntStream.range(0, getSize()).mapToObj(FlyRow::new);
  }

  @Override
  public IntFunction<FlyRow> getIntToRowFunction() {
    return FlyRow::new;
  }

  public static Builder builder(Metadata.Table table) {
    return new Builder(table);
  }

  public int getSize() {
    return size;
  }

  public static class Builder implements Consumer<Metadata.Row> {

    private final Metadata.Table table;
    private final List<StringColumn.Builder> stringBuilder = new ArrayList<>();
    private final List<DoubleColumn.Builder> numberBuilder = new ArrayList<>();
    private int size;

    public Builder(Metadata.Table table) {
      this.table = table;
      for (Metadata.Column column : table.getColumnList()) {
         if (column.getType() == Metadata.Column.Type.STRING) {
           numberBuilder.add(null);
           stringBuilder.add(StringColumn.builder());
         } else {
           numberBuilder.add(DoubleColumn.builder());
           stringBuilder.add(null);
         }
      }
    }

    public Chunk build() {
      List<Column<FlyRow>> columnList = new ArrayList<>();
      for (int i = 0; i < table.getColumnCount(); i++) {
        if (table.getColumn(i).getType() == Metadata.Column.Type.STRING) {
          columnList.add(stringBuilder.get(i).build());
        } else {
          columnList.add(numberBuilder.get(i).build());
        }
      }
      return new Chunk(size, columnList);
    }

    public void accept(Metadata.Row row) {
      size += 1;
      for (int i = row.getCellCount(); --i >= 0; ) {
        Metadata.Cell cell = row.getCell(i);
        if (table.getColumn(i).getType() == Metadata.Column.Type.STRING) {
          stringBuilder.get(i).accept(cell.getStringValueBytes().toByteArray());
        } else {
          numberBuilder.get(i)
              .accept(!cell.hasLongValue() ? (double) cell.getLongValue() : cell.getDoubleValue());
        }
      }
    }
  }

  public static class StringChunkBuilder implements Consumer<List<String>> {

    private final List<StringColumn.Builder> builderList;
    private int size;

    public StringChunkBuilder(Metadata.Table table) {
      for (Metadata.Column column : table.getColumnList()) {
        Preconditions.checkArgument(column.getType() == Metadata.Column.Type.STRING);
      }
      builderList = Stream.generate(StringColumn::builder)
          .limit(table.getColumnCount())
          .collect(Collectors.toList());
    }

    public void accept(List<String> strings) {
      size +=1;
      for (int i = 0; i < strings.size(); i++) {
        builderList.get(i).accept(strings.get(i).getBytes(StandardCharsets.UTF_8));
      }
    }

    public Chunk build() {
      List<Column<FlyRow>> columns = builderList.parallelStream()
          .map(StringColumn.Builder::build)
          .collect(Collectors.toList());
      return new Chunk(size, columns);
    }
  }
}
